/* =====================================================================================
 *       Filename:  queue.c
 *         Author:  MIEN
 *    Description:  队列实现(非线程安全)
 * =====================================================================================*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/stat.h>
#include "config.h"
#include "queue.h"

static char _sign[32];
static char _path[1024];

/* 以时间获取缓冲区标记 */
static void  _generateSign(int tick, char *sign)
{
    struct timeval tv;
    gettimeofday(&tv, NULL);
    sprintf(sign, "%lx-%x-%04x", tv.tv_sec, tv.tv_usec, tick);
}

static void _init_queue_path(const char *queue_name)
{
    char *data_dir = getConfig("queue_data_dir");
    sprintf(_path, "%s/%s", data_dir, queue_name);
}

static void _make_queue_dir(const char *queue_name)
{
    _init_queue_path(queue_name);
    struct stat st;
    if (stat(_path, &st) != -1)
       return;
    mkdir(_path, 0700);
}

/* 创建队列 */
queue *queueCreate(const char *queue_name)
{
    queue *q = malloc(sizeof(queue));

    strncpy(q->name, queue_name, sizeof(q->name));
    q->name[sizeof(q->name) - 1] = '\0';

    q->tick = 0;
    q->size = 0;

    q->write_offset = 0;
    q->read_offset  = 0;

    _generateSign(q->tick, q->write_sign);
    strcpy(q->read_sign, q->write_sign);

    q->read_cache[0] = '\0';

    q->suber = NULL;

    return q;
}

/* 释放队列 */
void queueFree(queue *q)
{
    subscriber *t, *s = q->suber;
    while (s != NULL) {
        t = s;
        s = s->next;
        free(t);
    }
    free(q);
}

/* 入队 */
int queueIn(queue *q, const char *data)
{
    int len = strlen(data);

    /* 数据超出缓冲区长度，返回错误 */
    if (len + 1 > sizeof(q->write_cache))
        return -2;

    /* 缓冲区不足以添加当前数据 */
    if (q->write_offset + len + 1 > sizeof(q->write_cache)) {

        /* 缓冲区已满，准备存储到文件  */
        _make_queue_dir(q->name);
        strcat(_path, "/");
        strcat(_path, q->write_sign);

        /* 建立文件 */
        FILE *fp = fopen(_path, "w");
        if (fp == NULL) return -1;

        /* 首先写入下一批数据的缓冲区标志 */
        _generateSign(q->tick, _sign);
        fwrite(_sign, 1, strlen(_sign), fp);
        fwrite("\0\n", 1, 2, fp);

        /* 写入队列数据 */
        fwrite(q->write_cache, 1, q->write_offset, fp);
        fclose(fp);

        /* 改变缓冲标志 */
        strcpy(q->write_sign, _sign);

        /* 重置写入缓冲区 */
        q->write_offset = 0;
    }

    /* 复制数据到缓冲区 */
    strncpy(q->write_cache + q->write_offset, data, len + 1);
    q->write_offset += len + 1;
    q->size++;
    q->tick = (q->tick + 1) & 0xFFFF;

    /* 返回成功 */
    return 0;
}

/* 出队*/
char *queueOut(queue *q)
{
    /* 队列长度为0 */
    if (q->size == 0) return NULL;

    /* 缓冲区大小 */
    int cache_size = sizeof(q->read_cache);

    /* 如果读取缓冲区元素已全部出队,
     * 从写入缓冲区，或者缓存的文件中读取 */
    if (q->read_offset >= cache_size || q->read_cache[q->read_offset] == '\0') {
        if (strcmp(q->read_sign, q->write_sign) == 0) {
            memcpy(q->read_cache, q->write_cache, q->write_offset);
            q->read_cache[q->write_offset] = '\0';
            q->write_offset = 0;
            q->read_offset = 0;
        }
        else {
            _init_queue_path(q->name);
            strcat(_path, "/");
            strcat(_path, q->read_sign);

            FILE *fp = fopen(_path, "r");
            if (!fp) return NULL;
            if (!fgets(_sign, sizeof(_sign), fp))
                return NULL;
            strcpy(q->read_sign, _sign);

            int read_bytes = 0;
            int n, read_limit;
            while (read_bytes < cache_size) {
                read_limit = cache_size - read_bytes;
                if (read_limit > 4096) read_limit = 4096;
                n = fread(q->read_cache + read_bytes, 1, read_limit, fp);
                if (n <= 0) break;
                read_bytes += n;
            }

            if (read_bytes < cache_size)
                q->read_cache[read_bytes] = '\0';

            fclose(fp);
            unlink(_path);

            q->read_offset = 0;
        }
    }

    /* 返回一项队列数据 */
    char *d = q->read_cache + q->read_offset;
    q->read_offset += strlen(d) + 1;
    q->size--;
    return d;
}

/* 订阅 */
int queueSub(queue *q, char *url, int number, int interval)
{
    /* 遍历订阅者 */
    subscriber *t = NULL, *n = NULL, *s = q->suber;
    while (s != NULL) {
        t = s;
        if (strncmp(t->url, url, sizeof(t->url)) == 0) {
            n = t;
            break;
        }
        s = s->next;
    }

    /* 没有发现相同url的订阅 */
    if (n == NULL) {

        /* 创建订阅者 */
        n = malloc(sizeof(subscriber));
        strncpy(n->url, url, sizeof(n->url));
        n->url[sizeof(n->url) - 1] = '\0';
        n->next = NULL;

        /* 之前没有订阅者 */
        if (t == NULL) {
            q->suber = n;
        } else {
            t->next = n;
        }
    }

    /* 更新订阅参数 */
    n->number = number;
    n->interval = interval;
    time(&(n->tm));

    return 1;
}

/* 退订 */
int queueUnsub(queue *q, char *url)
{
    subscriber *prev, *curr;
    prev = NULL;
    curr = q->suber;

    while (curr != NULL) {
        if (strncmp(curr->url, url, sizeof(curr->url)) == 0) {
            if (prev == NULL) {
                q->suber = curr->next;
            } else {
                prev->next = curr->next;
            }

            free(curr);
            return 1;
        }
        prev = curr;
        curr = curr->next;
    }
    return 0;
}

/* 持久化存储队列 */
void queueDump(queue *q)
{
    _make_queue_dir(q->name);
    strcat(_path, "/_dump");

    FILE *fp = fopen(_path, "w");
    if (fp == NULL) return;

    int len = strlen(q->name) + 1;
    fwrite(&len, 1, sizeof(int), fp);
    fwrite(q->name, 1, len, fp);

    len = strlen(q->write_sign) + 1;
    fwrite(&len, 1, sizeof(int), fp);
    fwrite(q->write_sign, 1, len, fp);

    len = strlen(q->read_sign) + 1;
    fwrite(&len, 1, sizeof(int), fp);
    fwrite(q->read_sign, 1, len, fp);

    fwrite(&(q->tick), 1, sizeof(int), fp);
    fwrite(&(q->size), 1, sizeof(int), fp);
    fwrite(&(q->write_offset), 1, sizeof(int), fp);

    fwrite(q->write_cache, 1, q->write_offset, fp);

    int cache_size = sizeof(q->read_cache);
    char *p = q->read_cache + q->read_offset;
    char *z = p;
    while (z < (q->read_cache + cache_size) && (*z) != '\0') {
        z += strlen(z) + 1;
    }
    len = z - p;
    fwrite(&len, 1, sizeof(int), fp);
    fwrite(p, 1, len, fp);

    /* 保存订阅者 */
    len = 0;
    subscriber *s = q->suber;
    while (s != NULL) {
        len++;
        s = s->next;
    }
    fwrite(&len, 1, sizeof(int), fp);

    s = q->suber;
    while (s != NULL) {
        len = strlen(s->url) + 1;
        fwrite(&len, 1, sizeof(int), fp);
        fwrite(s->url, 1, len, fp);
        fwrite(&s->number, 1, sizeof(int), fp);
        fwrite(&s->interval, 1, sizeof(int), fp);
        s = s->next;
    }
    
    // 关闭文件指针
    fclose(fp);
}

/* 从磁盘文件恢复队列 */
void queueRestore(queue *q)
{
    _init_queue_path(q->name);
    strcat(_path, "/_dump");

    FILE *fp = fopen(_path, "r");
    if (fp == NULL) return;
    
    int len;
    fread(&len, 1, sizeof(int), fp);
    fread(q->name, 1, len, fp);

    fread(&len, 1, sizeof(int), fp);
    fread(q->write_sign, 1, len, fp);

    fread(&len, 1, sizeof(int), fp);
    fread(q->read_sign, 1, len, fp);

    fread(&(q->tick), 1, sizeof(int), fp);
    fread(&(q->size), 1, sizeof(int), fp);
    fread(&(q->write_offset), 1, sizeof(int), fp);

    fread(q->write_cache, 1, q->write_offset, fp);

    fread(&len, 1, sizeof(int), fp);
    fread(q->read_cache, 1, len, fp);
    q->read_cache[len] = '\0';
    q->read_offset = 0;

    /* 恢复订阅者 */
    subscriber *s, *prev = NULL;
    q->suber = NULL;
    
    int c;
    fread(&c, 1, sizeof(int), fp);

    while (c > 0) {
        s = malloc(sizeof(subscriber));

        fread(&len, 1, sizeof(int), fp);
        fread(s->url, 1, len, fp);
        fread(&(s->number), 1, sizeof(int), fp);
        fread(&(s->interval), 1, sizeof(int), fp);
        time(&(s->tm));

        if (prev == NULL) {
            q->suber = s;
        } else {
            prev->next = s;
        }

        prev = s;
        c--;
    }

    // 关闭并删除文件
    fclose(fp);
    unlink(_path);
}

